package main.scala.demo

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * SparkReadJDBC
  *
  * @author zhangyimin
  * @date 2018-10-10 上午10:04
  * @version 1.0
  */
object SparkReadJDBC {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("mySales")
    conf.setMaster("local")
    //    val sc=new SparkContext(conf)
    val spark = SparkSession.builder().config(conf).getOrCreate()

    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/hive_etl?user=root&password=123456")
      //必须写表名
      .option("dbtable", "emp")
      .load()


//    println(jdbcDF.show())


    //查询数据库中的id, name, telephone三个列并以parquet（列存储）的方式存储在src/main/resources/sec_users路径下（存储后记得将名字改为user.parquet）
    jdbcDF.select("empno", "ename", "sal").write.format("parquet").save("src/main/resources/sec_users_parquet"+(Math.random()*100).intValue())
    //查询数据库中的username, name, telephone三个列并以parquet（列存储）的方式存储在src/main/resources/sec_users路径下存储后记得将名字改为user.json）
    jdbcDF.select("empno", "ename", "hiredate").write.format("json").save("src/main/resources/sec_users_json"+(Math.random()*100).intValue())

//    jdbcDF.filter("sal".lt(20)).show


    //存储成为一张虚表user_abel
    jdbcDF.select("empno", "ename", "job").write.mode("overwrite").saveAsTable("user_emp")
    val jdbcSQl = spark.sql("select * from user_emp where ename like 'S%' ")
//    jdbcSQl.as[emp].show
    jdbcSQl.show()

    jdbcSQl.write.format("json").save("./out/resulted_json"+(Math.random()*100).intValue())










  }

  case class emp(ename:String)

}
